package com.atguigu.flink.cep;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2023/2/3
 *
 *  超时(TimeOut): 指在模式中可以定义一个时间范围属性，可以约束下一条数据到达的时限。
 *
 *        规则1.within(Time.seconds(2)).next(规则2):
 *              数据必须以规则1开头(id=s1)，这条数据的下一条数据，必须在 2000ms以内(<2000ms)到达.
 *                  如果按点到了，就没有超时。
 *                  否则，就是超时。
 *
 *                  三种情况:
 *                          ①超时的情况
 *                                  超时的数据可以获取！
 *                          ②未超时
 *                                  匹配成功  可以获取!
 *                                  匹配失败  无法获取
 *
 *           超时的定义:
 *                      当前数据为s1,下一条数据间隔2s或更大时间才到达，当前数据超时!
 *
 */
public class Demo9_TimeOut
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((e, r) -> e.getTs());


        SingleOutputStreamOperator<WaterSensor> ds =
            env.readTextFile("data/ceptimeout.txt")
               .map(new WaterSensorMapFunction())
               .assignTimestampsAndWatermarks(watermarkStrategy);


        Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("规则1")
            .where(new SimpleCondition<WaterSensor>()
            {
                //定义匹配规则
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "s1".equals(value.getId());
                }
            })
            .within(Time.seconds(2))
            .next("规则2")
            .where(new SimpleCondition<WaterSensor>()
            {
                //定义匹配规则
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "s2".equals(value.getId());
                }
            });


        PatternStream<WaterSensor> patternStream = CEP.pattern(ds, pattern);

        OutputTag<String> outputTag = new OutputTag<>("late", TypeInformation.of(String.class));


        SingleOutputStreamOperator<String> ds1 = patternStream.select(
            outputTag
            ,
            new PatternTimeoutFunction<WaterSensor, String>()
            {

                @Override
                public String timeout(Map<String, List<WaterSensor>> pattern, long timeoutTimestamp) throws Exception {
                    return pattern.toString();
                }
            },
            new PatternSelectFunction<WaterSensor, String>()
            {

                //匹配成功
                @Override
                public String select(Map<String, List<WaterSensor>> map) throws Exception {
                    return map.toString();
                }
            }
        );
        //获取正常匹配的数据
        ds1.print().setParallelism(1);
        //获取超时的数据
        ds1.getSideOutput(outputTag).printToErr().setParallelism(1);


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
